package drds.binlog.database_driver;

import drds.binlog.common.Authors;
import drds.binlog.database_driver.packets.HeaderPacket;
import drds.binlog.database_driver.packets.client.ClientAuthenticationPacket;
import drds.binlog.database_driver.packets.client.QuitCommandPacket;
import drds.binlog.database_driver.packets.server.ErrorPacket;
import drds.binlog.database_driver.packets.server.HandshakeInitializationPacket;
import drds.binlog.database_driver.packets.server.Reply323Packet;
import drds.binlog.database_driver.socket.SocketChannel;
import drds.binlog.database_driver.socket.SocketChannels;
import drds.binlog.database_driver.utils.Packets;
import drds.binlog.database_driver.utils.PasswordEncrypter;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 基于socket协议实现
 */
@Slf4j
@Author(name = Authors.LI_YANG)
public class Connector
{

    public static final int timeout = 5 * 1000;                                     // 5s


    @Setter
    @Getter
    private SocketAddress socketAddress;
    @Setter
    @Getter
    private String username;
    @Setter
    @Getter
    private String password;
    @Setter
    @Getter
    private byte charsetNumber = 33;
    @Setter
    @Getter
    private String defaultSchema = "test";
    @Setter
    @Getter
    private int soTimeout = 30 * 1000;
    @Setter
    @Getter
    private int connTimeout = 5 * 1000;
    @Setter
    @Getter
    private int receiveBufferSize = 16 * 1024;
    @Setter
    @Getter
    private int sendBufferSize = 16 * 1024;
    @Setter
    @Getter
    private SocketChannel socketChannel;
    @Setter
    @Getter
    private volatile boolean dumping = false;
    // mysql connectinnId
    @Setter
    @Getter
    private long connectionId = -1;
    private AtomicBoolean connected = new AtomicBoolean(false);

    public Connector()
    {
    }

    public Connector(InetSocketAddress inetSocketAddress, String username, String password)
    {
        String hostString = inetSocketAddress.getHostString();
        int port = inetSocketAddress.getPort();
        this.socketAddress = new InetSocketAddress(hostString, port);

        this.username = username;
        this.password = password;
    }

    public Connector(InetSocketAddress inetSocketAddress, String defaultSchema, String username, String password, byte charsetNumber)
    {
        this(inetSocketAddress, username, password);

        this.charsetNumber = charsetNumber;
        this.defaultSchema = defaultSchema;
    }

    public void connect() throws IOException
    {
        if (connected.compareAndSet(false, true))
        {
            try
            {
                socketChannel = SocketChannels.open(socketAddress);
                negotiate(socketChannel);
            } catch (Exception e)
            {
                disconnect();
                throw new IOException("connect " + this.socketAddress + " failure", e);
            }
        } else
        {
            log.error("the socketChannel can't be connected twice.");
        }
    }

    public void reconnect() throws IOException
    {
        disconnect();
        connect();
    }

    public void disconnect() throws IOException
    {
        if (connected.compareAndSet(true, false))
        {
            try
            {
                if (socketChannel != null)
                {
                    socketChannel.close();
                }
                log.info("disConnect MysqlConnection to {}...", socketAddress);
            } catch (Exception e)
            {
                throw new IOException("disconnect " + this.socketAddress + " failure", e);
            }

            // 执行一次quit
            if (dumping && connectionId >= 0)
            {
                Connector connector = null;
                try
                {
                    connector = this.fork();
                    connector.connect();
                    UpdateExecutor executor = new UpdateExecutor(connector);
                    executor.update("KILL CONNECTION " + connectionId);
                } catch (Exception e)
                {
                    // 忽略具体异常
                    log.info("KILL DUMP " + connectionId + " failure", e);
                } finally
                {
                    if (connector != null)
                    {
                        connector.disconnect();
                    }
                }

                dumping = false;
            }
        } else
        {
            log.info("the socketChannel {} is not connected", this.socketAddress);
        }
    }

    public boolean isConnected()
    {
        return this.socketChannel != null && this.socketChannel.isConnected();
    }

    public Connector fork()
    {
        Connector connector = new Connector();
        connector.setCharsetNumber(getCharsetNumber());
        connector.setDefaultSchema(getDefaultSchema());
        connector.setSocketAddress(getSocketAddress());
        connector.setPassword(password);
        connector.setUsername(getUsername());
        connector.setReceiveBufferSize(getReceiveBufferSize());
        connector.setSendBufferSize(getSendBufferSize());
        connector.setSoTimeout(getSoTimeout());
        connector.setConnTimeout(connTimeout);
        return connector;
    }

    public void quit() throws IOException
    {
        QuitCommandPacket quitCommandPacket = new QuitCommandPacket();
        byte[] cmdBody = quitCommandPacket.toBytes();

        HeaderPacket quitHeader = new HeaderPacket();
        quitHeader.setPacketBodyLength(cmdBody.length);
        quitHeader.setPacketSequenceNumber((byte) 0x00);
        Packets.write(socketChannel, quitHeader.toBytes(), cmdBody);
    }

    private void negotiate(SocketChannel channel) throws IOException
    {
        HeaderPacket header = Packets.readHeader(channel, 4, timeout);
        byte[] body = Packets.readBytes(channel, header.getPacketBodyLength(), timeout);
        if (body[0] < 0)
        {// check field_count
            if (body[0] == -1)
            {
                ErrorPacket error = new ErrorPacket();
                error.fromBytes(body);
                throw new IOException("handshake exception:\n" + error.toString());
            } else if (body[0] == -2)
            {
                throw new IOException("Unexpected EOF packet at handshake phase.");
            } else
            {
                throw new IOException("unpexpected packet with field_count=" + body[0]);
            }
        }
        HandshakeInitializationPacket handshakePacket = new HandshakeInitializationPacket();
        handshakePacket.fromBytes(body);
        connectionId = handshakePacket.threadId; // 记录一下connection

        log.info("handshake initialization packet received, prepare the client authentication packet to send");

        ClientAuthenticationPacket clientAuth = new ClientAuthenticationPacket();
        clientAuth.setCharsetNumber(charsetNumber);

        clientAuth.setUsername(username);
        clientAuth.setPassword(password);
        clientAuth.setServerCapabilities(handshakePacket.serverCapabilities);
        clientAuth.setDatabaseName(defaultSchema);
        clientAuth.setScrumbleBuff(joinAndCreateScrumbleBuff(handshakePacket));

        byte[] clientAuthPkgBody = clientAuth.toBytes();
        HeaderPacket h = new HeaderPacket();
        h.setPacketBodyLength(clientAuthPkgBody.length);
        h.setPacketSequenceNumber((byte) (header.getPacketSequenceNumber() + 1));

        Packets.write(channel, h.toBytes(), clientAuthPkgBody);
        log.info("client authentication packet is sent out.");

        // check auth result
        header = null;
        header = Packets.readHeader(channel, 4);
        body = null;
        body = Packets.readBytes(channel, header.getPacketBodyLength(), timeout);
        assert body != null;
        if (body[0] < 0)
        {
            if (body[0] == -1)
            {
                ErrorPacket err = new ErrorPacket();
                err.fromBytes(body);
                throw new IOException("Error When doing Client Authentication:" + err.toString());
            } else if (body[0] == -2)
            {
                auth323(channel, header.getPacketSequenceNumber(), handshakePacket.seed);
                // throw new
                // IOException("Unexpected EOF packet at Client Authentication.");
            } else
            {
                throw new IOException("unpexpected packet with field_count=" + body[0]);
            }
        }
    }

    private void auth323(SocketChannel socketChannel, byte packetSequenceNumber, byte[] seed) throws IOException
    {
        // auth 323
        Reply323Packet reply323Packet = new Reply323Packet();
        if (password != null && password.length() > 0)
        {
            reply323Packet.seed = PasswordEncrypter.scramble323(password, new String(seed)).getBytes();
        }
        byte[] b323Body = reply323Packet.toBytes();

        HeaderPacket h323 = new HeaderPacket();
        h323.setPacketBodyLength(b323Body.length);
        h323.setPacketSequenceNumber((byte) (packetSequenceNumber + 1));

        Packets.write(socketChannel, h323.toBytes(), b323Body);
        log.info("client 323 authentication packet is sent out.");
        // check auth result
        HeaderPacket headerPacket = Packets.readHeader(socketChannel, 4);
        byte[] body = Packets.readBytes(socketChannel, headerPacket.getPacketBodyLength());
        assert body != null;
        switch (body[0])
        {
            case 0:
                break;
            case -1:
                ErrorPacket err = new ErrorPacket();
                err.fromBytes(body);
                throw new IOException("Error When doing Client Authentication:" + err.toString());
            default:
                throw new IOException("unpexpected packet with field_count=" + body[0]);
        }
    }

    private byte[] joinAndCreateScrumbleBuff(HandshakeInitializationPacket handshakeInitializationPacket) throws IOException
    {
        byte[] dest = new byte[handshakeInitializationPacket.seed.length + handshakeInitializationPacket.restOfScrambleBuff.length];
        System.arraycopy(handshakeInitializationPacket.seed, 0, dest, 0, handshakeInitializationPacket.seed.length);
        System.arraycopy(handshakeInitializationPacket.restOfScrambleBuff,
                0,
                dest,
                handshakeInitializationPacket.seed.length,
                handshakeInitializationPacket.restOfScrambleBuff.length);
        return dest;
    }


}
